package ru.unlocker.reports.domain;

import java.time.LocalDate;
import java.time.temporal.ChronoUnit;
import java.util.UUID;
import java.util.stream.Stream;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class ReportEventProcessor extends KeyedProcessFunction<UUID, RawReportEvent, ReportAndTimerEvent> {

  private ValueState<Long> bestBeforeTime;
  private ValueState<LocalDate> workday;
  private ValueState<String> reportCode;

  @Override
  public void processElement(
      RawReportEvent event,
      KeyedProcessFunction<UUID, RawReportEvent, ReportAndTimerEvent>.Context ctx,
      Collector<ReportAndTimerEvent> out
  ) throws Exception {
    switch (event.getReportEvent()) {
      case REPORT_STARTED -> {
        workday.update(event.getWorkday());
        reportCode.update(event.getReportCode());
        // закладываем период удержания состояния 1 сутки
        long bestBeforeMilli = event.getCreatedAt().plus(1, ChronoUnit.DAYS).toEpochMilli();
        ctx.timerService().registerEventTimeTimer(bestBeforeMilli);
        bestBeforeTime.update(bestBeforeMilli);
        out.collect(
            ReportAndTimerEvent.builder()
                .createdAt(event.getCreatedAt())
                .reportEvent(ReportEventKind.REPORT_STARTED)
                .reportId(event.getReportId())
                .groupId(event.getGroupId())
                .workday(event.getWorkday())
                .reportCode(event.getReportCode())
                .build()
        );
      }
      case REPORT_FINISHED -> {
        String reportCodeStr = reportCode.value();
        // старое событие, состояние уже очищено
        if (reportCodeStr == null) {
          return;
        }
        out.collect(
            ReportAndTimerEvent.builder()
                .createdAt(event.getCreatedAt())
                .reportEvent(ReportEventKind.REPORT_FINISHED)
                .reportId(event.getReportId())
                .groupId(event.getGroupId())
                .workday(workday.value())
                .reportCode(reportCodeStr)
                .duration(event.getDuration())
                .build()
        );
        ctx.timerService().deleteEventTimeTimer(bestBeforeTime.value());
        clearState();
      }
      case REPORT_FAILED -> {
        String reportCodeStr = reportCode.value();
        // старое событие, состояние уже очищено
        if (reportCodeStr == null) {
          return;
        }
        out.collect(
            ReportAndTimerEvent.builder()
                .createdAt(event.getCreatedAt())
                .reportEvent(ReportEventKind.REPORT_FAILED)
                .reportId(event.getReportId())
                .groupId(event.getGroupId())
                .workday(workday.value())
                .reportCode(reportCodeStr)
                .errorMessage(event.getErrorMessage())
                .build()
        );
        ctx.timerService().deleteEventTimeTimer(bestBeforeTime.value());
        clearState();
      }
    }
  }

  @Override
  public void open(OpenContext openContext) {
    ValueStateDescriptor<Long> bestBeforeTimeDescriptor = new ValueStateDescriptor<>("bestBeforeTime", Long.class);
    ValueStateDescriptor<LocalDate> workdayDescriptor = new ValueStateDescriptor<>("workday", LocalDate.class);
    ValueStateDescriptor<String> reportCodeDescriptor = new ValueStateDescriptor<>("reportCode", String.class);
    bestBeforeTime = getRuntimeContext().getState(bestBeforeTimeDescriptor);
    workday = getRuntimeContext().getState(workdayDescriptor);
    reportCode = getRuntimeContext().getState(reportCodeDescriptor);
  }

  @Override
  public void onTimer(
      long timestamp,
      KeyedProcessFunction<UUID, RawReportEvent, ReportAndTimerEvent>.OnTimerContext ctx,
      Collector<ReportAndTimerEvent> out
  ) {
    clearState();
  }

  private void clearState() {
    Stream.of(bestBeforeTime, workday, reportCode).forEach(ValueState::clear);
  }
}
